
#include "Debug.h"
#include <sstream>
#include <string>
#include <stdlib.h>
#include <sys/stat.h>
#include <libpq-fe.h>
#include <map>
#include <string.h>
#include <glib.h>
#include <assert.h>
#include "config.h"
#include "define.h"
#include "CobwebSqlite.h"

using namespace std;

extern "C" {
  bool nidConsInit(char *consumer, const char* conf);
  int nidConsFriends(char *user, GHashTable *frs);
  int nidConsUpdate(char *user, long expire, GHashTable *frs);
  void nidConsShutdown();
  void nidConsFlush(char *consumer);
};

PGconn *g_db = NULL;
CobWebSqlite *pDB = NULL;
static int g_expire = 86400*3;
bool nidConsInit(char *consumer, const char* conf)
{
        struct stat st;
        if (lstat(conf, &st) != 0)
        {
                ERRO<<":"<<conf<<" is not exist!"<<endl;
                return false;
        }
        Configure cfg;
        map <string, string > g_Confs;
        g_Confs = cfg.LoadConfInfo(conf);
        cfg.Print();
        if(g_Confs.size() <= 0)
        {
                cerr<<"error to explain conf :"<< conf<<endl;
                return true;
        }
        map <string, string >::iterator iter;
        string dbhost, dbuser, dbpass, dbname;
        if (g_Confs.find("dbhost") != g_Confs.end())
                dbhost = g_Confs["dbhost"];
        if (g_Confs.find("dbuser") != g_Confs.end())
                dbuser = g_Confs["dbuser"];
        if (g_Confs.find("dbpass") != g_Confs.end())
                dbpass = g_Confs["dbpass"];
        if (g_Confs.find("dbname") != g_Confs.end())
		dbname = g_Confs["dbname"];
        string dbport = "5432";
        if (g_Confs.find("dbport") != g_Confs.end())
		dbport = g_Confs["dbport"];

	g_db = PQsetdbLogin(dbhost.c_str(), dbport.c_str(), NULL, NULL, dbname.c_str(), dbuser.c_str(), dbpass.c_str());

	if(PQstatus(g_db) != CONNECTION_OK)
	{
		ERRO << PQerrorMessage(g_db) << endl;
		PQfinish( g_db);
		g_db = NULL;
                return false;
	}
        string smax, dbfile;
        if (g_Confs.find("maxtable") != g_Confs.end())
                smax = g_Confs["maxtable"];

        if (g_Confs.find("dbfile") != g_Confs.end())
		dbfile = g_Confs["dbfile"];
        int imaxt = atoi(smax.c_str());
        if(imaxt < 1)
           imaxt = 1;

        pDB = new CobWebSqlite(dbfile.c_str(), imaxt);
        if(!pDB->Initialize())
             return false;

        string s; 
        if (g_Confs.find("uexpire") != g_Confs.end())
                s = g_Confs["uexpire"];
        g_expire = atoi(s.c_str()); 

        return true;
}

int nidConsFriends(char * user, GHashTable *frs)
{
   string val, name;
   long  expire = 0;
   pDB->Fetch(user, name , val, expire);
   if(val.length()<= 0)
   {
	   if(name.length() > 0)
		   return 1;
	   DEBUG1("fetch from SQLITE, no data!");
	   return 0;
   } 
   // val = stamp:name;nid:name;nid:name;nid:name
   char **columns = g_strsplit(val.c_str(), ";", -1);
   int columns_num =  g_strv_length(columns);
   if(columns_num < 2)
   {
           DEBUG1("data is error!");
	   g_strfreev(columns);
	   return 1;
   }
  
   FObject *uself = new FObject(user, 1, (char *)name.c_str()) ;
   g_hash_table_insert(frs, g_strdup(user), uself);

   for(int i=0; i< columns_num ; i++ )
   { // g_strstrip(columns[i]);
	   char **scolumns = g_strsplit(columns[i], ":", -1);
	   int cnum =  g_strv_length(scolumns);
	   if(cnum != 2)
	   {
		   g_strfreev(scolumns);
		   continue;
	   }

	   FObject *uself = new FObject(scolumns[0], 1, scolumns[1] ) ;
	   g_hash_table_insert(frs, g_strdup(scolumns[0]), uself);
	   g_strfreev(scolumns);
   }
   g_strfreev(columns);
   DEBUG2("YES, Data size = ", columns);
   return  columns_num;
}

int nidConsUpdate(char * user, long lt,  GHashTable *frs)
{
	assert(user != NULL);

	string val, name;
        long stamp = 0;
	pDB->Fetch(user, name , val, stamp);
	if(val.length() > 0  && stamp <= lt + g_expire )
	{ // val = stamp:name;nid:name;nid:name;nid:name
		
		FObject *uself = new FObject(user, 1, (char *)name.c_str()) ;
		g_hash_table_insert(frs, g_strdup(user), uself);
 char **columns = g_strsplit(val.c_str(), ";", -1);
   int columns_num =  g_strv_length(columns);

		for(int i=0; i< columns_num ; i++ )
		{ // g_strstrip(columns[i]);
			char **scolumns = g_strsplit(columns[i], ":", -1);
			int cnum =  g_strv_length(scolumns);
			if(cnum != 2)
			{
				g_strfreev(scolumns);
				continue;
			}

			FObject *uself = new FObject(scolumns[0], 1, scolumns[1] ) ;
			g_hash_table_insert(frs, g_strdup(scolumns[0]), uself);
			g_strfreev(scolumns);
		}
		g_strfreev(columns);
		return  columns_num;
	}

	if(g_db == NULL)
	{
		cerr<<"db connection is NULL"<<endl;
		return -1;
	}

	int u = atoi(user);
	stringstream sql;
	sql <<"select nick from user_profile where id="<<u;
	PGresult *result = PQexec(g_db, sql.str().c_str());
	if(PQresultStatus(result) != PGRES_TUPLES_OK)
	{
		cerr << PQerrorMessage(g_db) << endl;
		PQclear(result);
		return -1;
	}
	long rownum = PQntuples(result);
	if (rownum <= 0 )
	{
		PQclear(result);
		return 0;
	}
        string sval;
	FObject *uself = new FObject(user, 1,  (char *)PQgetvalue(result, 0, 0) ) ;
	g_hash_table_insert(frs, g_strdup(user), uself);
        sql.str("");

        long expire = lt + g_expire;
        name = (char *)PQgetvalue(result, 0, 0);       

	PQclear(result);
	sql.str(""); 
	sql<<"select u.partnerid, p.nick from user_relation as u, user_profile as p where ownerid ="<<u<<" and class=1 and u.partnerid=p.id";
	DEBUG1(sql.str());
	result = PQexec(g_db, sql.str().c_str());
	if(PQresultStatus(result) != PGRES_TUPLES_OK)
	{
		cerr << PQerrorMessage(g_db) << endl;
		PQclear(result);
		return -1;
	}

	rownum = PQntuples(result);
	DEBUG2("fetch row = ", rownum);
	if (rownum == 0 )
	{
		PQclear(result);
		return 0;
	}
	for (int i = 0; i < rownum; i++)
	{
		FObject *uf = new FObject(PQgetvalue(result,i,0), 1,  (char *)PQgetvalue(result, i, 1) ) ;
                sql.str("");
                if(i > 0)
                   sql<<";";
		sql<< (char *)PQgetvalue(result, i, 0) <<":"<< (char *)PQgetvalue(result, i, 1) ;       
		sval.append(sql.str());
		g_hash_table_insert(frs, g_strdup(PQgetvalue(result, i, 0)), uf);
	}
	PQclear(result);
        DEBUG1(sval.c_str() );
	if( stamp > 0)
		pDB->Replace(user,(char *)name.c_str(), (char *) sval.c_str(), expire);
	else
		pDB->Save(user, (char *)name.c_str(), (char *) sval.c_str(), expire);
	return rownum;
}

void nidConsShutdown()
{
	if(g_db != NULL){
		PQfinish( g_db);
		g_db = NULL;
	}
        if(pDB != NULL)
           delete pDB; 
 // Do whatever to close/flush/sync the consumer's work before exit...
  //   delete gx_TxLog;
}

void nidConsFlush(char * consumer)
{
  // If this consumer keeps statistics or writes to a tx log, then
  // flush the statistics here, otherwise do nothing.
}


